package com.chencong.transform;

import com.chencong.env.FlinkTableEnv;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author chencong
 * @Description 对两个或者两个以上的DataStream进行union操作，产生一个包含所有DataStream元素的新DataStream
 * @Date 7:48 下午 2021/8/18
 * @Param
 **/
public class Union {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamTableEnvironment = FlinkTableEnv.getStreamTableEnvironment();
        streamTableEnvironment.setParallelism(1);

        DataStreamSource<Integer> stream1 = streamTableEnvironment.fromElements(1, 2, 3, 4, 5);
        DataStreamSource<Integer> stream2 = streamTableEnvironment.fromElements(10, 20, 30, 40, 50);
        DataStreamSource<Integer> stream3 = streamTableEnvironment.fromElements(100, 200, 300, 400, 500);

        // 把多个流union在一起成为一个流, 这些流中存储的数据类型必须一样: 水乳交融
        stream1
                .union(stream2)
                .union(stream3)
                .print("三次union");

        streamTableEnvironment.execute();

    }
}
